package com.whale.springboot.kafka.java.producer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducerCallbackPartitions {

    public static void main(String[] args) throws InterruptedException {

        // 0 配置
        Properties properties = new Properties();

        // 2. 给 kafka 配置对象添加配置信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.179.131:9092,192.168.179.131:9093,192.168.179.131:9094");
        // key,value 序列化（必须）： key.serializer， value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 关联自定义分区器
//        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.whale.kafka.java.producer.MyPartitioner");

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        /**
         * 采用了异步方式 去发送消息
         */
        for (int i = 0; i < 5; i++) {
            // 指定数据发送到 1 号分区， key 为空（IDEA 中 ctrl + p 查看参数）
            kafkaProducer.send(new ProducerRecord<>("jkhtest002", 0, "","JKHJKHJKH " + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata,
                                         Exception e) {
                    if (e == null){
                        System.out.println(" 主 题 ： " +
                                metadata.topic() + "->" + "分区： " + metadata.partition()
                        );
                    }else {
                        e.printStackTrace();
                    }
                }
            });
        }
        kafkaProducer.close();
    }
}
